package com.air.antispider.stream.rulecompute.businessprocess

import java.util
import com.air.antispider.stream.commom.bean.ProcessedData
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.rdd.RDD

import scala.collection.mutable.ArrayBuffer

object RuleComputeUtils {
  /**
    * 某个 IP，5 分钟内关键页面的访问次数的 Cookie 数
    * @param processedDataRDD
    * @param pages
    * @return
    */
  def ipCookieCount(processedDataRDD: RDD[ProcessedData], pages: ArrayBuffer[String]): collection.Map[String, Int] = {
    processedDataRDD.filter(line => {
      val url: String = line.request
      var flag = 0
      for (page <- pages) {
        if (url.matches(page)){
          flag = 1
        }
      }
      flag == 1
    }).map(line => {
      val ip: String = line.remoteAddr
      //取出sessionID
      val sessionID: String = line.cookieValue_JSESSIONID
      (ip,sessionID)
    })
      .groupByKey()
      .map(line => {
        //去重存入map中
        val sessionIterator: Iterable[String] = line._2
        var set = Set[String]()
        for (sessionID <- sessionIterator) {
          set += sessionID
        }
        (line._1,set.size)
      }).collectAsMap()
  }

  /**
    * 某个 IP，5 分钟内查询不同行程的次数
    *
    * @param processedDataRDD
    * @return
    */
  def ipArrCount(processedDataRDD: RDD[ProcessedData]): collection.Map[String, Int] = {
    processedDataRDD.map(line => {
      val ip: String = line.remoteAddr
      val arrcity: String = line.requestParams.arrcity
      val depcity: String = line.requestParams.depcity
      (ip,(arrcity,depcity))
    }).groupByKey()
      .map(line => {
        //取出所有行程信息
        val cityIterator: Iterable[(String, String)] = line._2
        var set = Set[String]()
        for (citys <- cityIterator) {
          val city: String = citys._1 + "->" + citys._2
          set += city
        }
        //不同行程的次数
        (line._1,set.size)
      }).collectAsMap()
  }

  /**
    * 某个 IP，5 分钟内小于最短访问间隔（自设）的关键页面查询次数
    *
    * @param processedDataRDD
    * @param pages
    * @return
    */
  def ipCriticalPagesMinNumTime(processedDataRDD: RDD[ProcessedData], pages: ArrayBuffer[String]): collection.Map[(String, String), Int] = {
    //获取关键页面
    processedDataRDD.filter(processedData => {
      val url: String = processedData.request
      var flag = 0
      for (page <- pages) {
        if (url.matches(page)){
          flag = 1
        }
      }
      flag == 1
    }).map(line =>{
      //将数据转换为((ip,URL),time)的格式
      val ip: String = line.remoteAddr
      val url: String = line.request
      val iso860: String = line.timeIso8601
      val timeStamp: Long = FastDateFormat.getInstance("yyyy-MM-dd'T'HH:mm:ss").parse(iso860).getTime
      ((ip, url), timeStamp)
    }).groupByKey()
      .map(line => {
        //求出时间差
        val timeStampIterator: Iterable[Long] = line._2
        val timeStampArray = timeStampIterator.toArray
        util.Arrays.sort(timeStampArray)
        //定义一个差值的集合
        var tmp = ArrayBuffer[Long]()

        //求差值，存入集合
        for (i <- 0 until timeStampArray.length - 1) {
          val min = timeStampArray(i + 1) - timeStampArray(i)
          tmp += min
        }
        //判断是否小于预设值，若是，统计
        var count = 0
        for (min <- tmp) {
          if (min < 3){
            count = count + 1
          }
        }
        (line._1, count)
      }).collectAsMap()

  }

  /**
    * ip访问关键页面的最小时间差
    *
    * @param processedDataRDD
    * @param criticalPagesList
    * @return
    */
  def ipCriticalPagesMinTime(processedDataRDD: RDD[ProcessedData], criticalPagesList: ArrayBuffer[String]): collection.Map[String, Int] = {
    //首先过滤无关页面
    processedDataRDD.filter(processedData =>{
      val url: String = processedData.request
      var flag = 0
      //与广播变量中的规则一一匹配
      for (criticalPage <- criticalPagesList) {
        if (url.matches(criticalPage)){
          flag = 1
        }
      }
      flag == 1
    }).map(line => {
      //将数据转为（ip，时间戳）
      val ip: String = line.remoteAddr
      val iso860: String = line.timeIso8601
      //格式化iso数据格式2019-06-04T17:17:07+08:00 -> 2019-06-04 17:17:07
      val timeStr: String = iso860.replace("T"," ").replace("+08:00", "")
      //转为时间戳格式
      val timeStamp: Long = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss").parse(timeStr).getTime
      (ip,timeStamp)
    })
    //按照ip进行分组
        .groupByKey()
      .map(line => {
        val ip: String = line._1
        //对所有时间戳先进行排序
        val timeIterator: Iterable[Long] = line._2
        //存入array中方便排序
        val timeArr: Array[Long] = timeIterator.toArray
        util.Arrays.sort(timeArr)
        //求两两之间的时间差，存入Set集合
        var set: Set[Long] = Set[Long]()
        for (i <- 0 until timeArr.length -1) {
          //当前时间和下一条的时间差值
          val currentTime: Long = timeArr(i)
          val nextTime: Long = timeArr(i + 1)
          set += (nextTime - currentTime)
        }
        //将set集合转为数组再次排序。取第一条即为最小时间差
        val sortArr: Array[Long] = set.toArray
        util.Arrays.sort(sortArr)
        //返回
        (ip,sortArr(0).toInt)
      }).collectAsMap()

  }

  /**
    * 同一ip携带不同UA的个数
    *
    * @param processedDataRDD
    * @return
    */
  def ipUACount(processedDataRDD: RDD[ProcessedData]): collection.Map[String, Int] = {
    processedDataRDD.map(processedData =>{
      //取出每条数据的ip和ua
      val ip: String = processedData.remoteAddr
      val ua: String = processedData.httpUserAgent
      (ip,ua)
    })
      //按照ip进行分组聚合操作(ip,(ua1,ua2,ua1,ua3))
      .groupByKey()
      .map(data =>{
        //对ua的集合去重
        val ip = data._1
        val uaIterator: Iterable[String] = data._2
        var set: Set[String] = Set[String]()
        //将所有ua存入set集合中去重
        for (ua <- uaIterator) {
          set += ua
        }
        (ip,set.size)
      }).collectAsMap()

  }

  /**
    * 同一ip访问关键页面的数量
    * @param processedDataRDD
    * @param criticalList
    * @return
    */
  def ipCriticalPagesCount(processedDataRDD: RDD[ProcessedData], criticalList: ArrayBuffer[String]): collection.Map[String, Int] = {
    //过滤关键页面
    processedDataRDD.filter(processedData =>{
      val url: String = processedData.request
      //定义一个标记
      var flag = 0
      //url与广播变量中的规则一一匹配
      for (criticalPage <- criticalList) {
        if (url.matches(criticalPage)){
          //该url应被过滤掉
          flag = 1
        }
      }
      flag == 1
    }).map(processedData => {
      val addr: String = processedData.remoteAddr
      //非空判断
      if (StringUtils.isNotBlank(addr) && !addr.equalsIgnoreCase("null")){
        (addr,1)
      }else{
        ("",1)
      }
    }).reduceByKey(_ + _).collectAsMap()

  }

  /**
    * 对每个ip的访问总量统计
    *
    * @param processedDataRDD
    * @return
    */
  def ipCount(processedDataRDD: RDD[ProcessedData]): collection.Map[String, Int] = {
    processedDataRDD.map(processedData =>{
      //取出ip
      val ip: String = processedData.remoteAddr
      //非空判断，记做1
      if (StringUtils.isNoneBlank(ip) && !ip.equals("null")) {
        (ip,1)
      }else {
        ("",1)
      }
    }).reduceByKey(_ + _).collectAsMap()

  }

  /**
    * 计算ip段的访问量
    *
    * @param processedDataRDD
    * @return
    */
  def ipBlock(processedDataRDD: RDD[ProcessedData]): collection.Map[String, Int] = {
    val ipBlockRDD: RDD[(String, Int)] = processedDataRDD.map(processedData => {
      val ip: String = processedData.remoteAddr
      val ipArr: Array[String] = ip.split("\\.")
      if (ipArr.length > 1) {
        //取出IP段
        val ipBlock = ipArr(0) + "." + ipArr(1)
        //计数为1
        //(192.168, 1)
        (ipBlock, 1)
      } else {
        ("", 1)
      }
    })
    ipBlockRDD
    //聚合计算
    val reduceRDD: RDD[(String, Int)] = ipBlockRDD.reduceByKey((x, y) => x + y)
    val map: collection.Map[String, Int] = reduceRDD.collectAsMap()

    map
  }

}
